package worker

import (
	"context"
	"fmt"
	"github.com/mongodb/mongo-go-driver/mongo"
	"go_job/common"
	"time"
)

type LogSink struct {
	client         *mongo.Client
	logCollection  *mongo.Collection
	logChan        chan *common.JobLog
	autoCommitChan chan *common.LogBatch
}

var (
	G_logSink *LogSink
)

func InitLogSink() (err error) {
	var (
		client *mongo.Client
	)
	if client, err = mongo.Connect(context.TODO(), G_config.MongoPath); err != nil {
		return
	}

	G_logSink = &LogSink{
		client:         client,
		logCollection:  client.Database("cron").Collection("log"),
		logChan:        make(chan *common.JobLog, 1000),
		autoCommitChan: make(chan *common.LogBatch, 1000),
	}
	go G_logSink.writeLoop()
	fmt.Println("mongodb日志模块初始化完成")
	return
}

func (logSink *LogSink) writeLoop() {
	var (
		log          *common.JobLog
		logBatch     *common.LogBatch
		commitTimer  *time.Timer
		timeOutBatch *common.LogBatch
	)

	//涉及并发问题,可能会有bug
	for {
		select {
		case log = <-logSink.logChan:
			if logBatch == nil {
				logBatch = &common.LogBatch{}
				//超时自动提交
				commitTimer = time.AfterFunc(time.Duration(G_config.LogAutoCommitTimer)*time.Millisecond,
					func(batch *common.LogBatch) func() {
						//发出通知
						return func() {
							logSink.autoCommitChan <- logBatch
						}
					}(logBatch),
				)
			}
			logBatch.Logs = append(logBatch.Logs, log)
			if len(logBatch.Logs) >= G_config.LogMaxSize {
				logSink.logCollection.InsertMany(context.TODO(), logBatch.Logs)
				logBatch = nil
				commitTimer.Stop()
			}
		case timeOutBatch = <-logSink.autoCommitChan:
			if timeOutBatch != logBatch {
				//跳过已提交的
				continue
			}
			logSink.logCollection.InsertMany(context.TODO(), timeOutBatch.Logs)
			logBatch = nil

		}
	}
}

func (logSink *LogSink) PushToLogChan(log *common.JobLog) {
	logSink.logChan <- log
}
